package org.liang.code.rocketMQ;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;


public class 接收消息 {

    public static void main(String[] args) throws MQClientException {

        // 创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Con1");
        // 设置NameServer节点
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.setMessageModel(MessageModel.BROADCASTING);

         /* 订阅主题，
            consumer.subscribe包含两个参数：
            topic: 说明消费者从Broker订阅哪一个主题，这一项要与Provider保持一致。
            subExpression: 子表达式用于筛选tags。
            同一个主题下可以包含很多不同的tags，subExpression用于筛选符合条件的tags进行
            接收。
            例如：设置为*，则代表接收所有tags数据。
            例如：设置为2020S4，则Broker中只有tags=2020S4的消息会被接收，而2020S2就
            会被排除在外。
        */
        consumer.subscribe("TTmusic", "*");

        // 创建监听，当有新的消息监听程序会及时捕捉并加以处理。
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {

            for (MessageExt msg : msgs) {

                System.out.println("消费者获取数据:" + msg.getMsgId() + "==>" + new String(msg.getBody()));

            }
            //返回数据已接收标识
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        // 启动消费者，与Broker建立长连接，开始监听。
        consumer.start();
    }
}
